多线程八 Lock

发表于 | 分类于

前面我们可以使用synchronized关键字来实现线程之间的同步互斥,lock接口同样也是在JDK1.5中提出,同样是解决线程安全性问题的另一种解决方案,而且它更强大,更灵活本片博客介绍对其展开介绍;

Lock接口有如下几个实现类:

  • ReentrantLock–JDK实现的锁
  • ReentrantReadWritrLock.ReadLock
  • ReentrantReadWriteLock.WriteLock

打个例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public class Demo01 {
    private int i=0;

 Lock Mylock = new ReentrantLock();

  public int add() throws InterruptedException {
      try{
          Mylock.lock();
          i++;
          return i;
      } finally {
          Mylock.unlock();
      }

      }

如上代码 i++ 被 Mylock.lock()和 Mylock.unlock()围住,显示的获取和释放锁,具有同步性…

其中,ReentrantLock一个可重入的互斥锁 Lock，它具有与使用 synchronized 方法和语句所访问的隐式监视器锁相同的一些基本行为和语义，但功能更强大。

Lock 与 Synchronized 相比较,显而易见的就是,Lock需要显示的去获取锁,释放锁,比较繁琐,但是繁琐带来了更大好处,让代码更灵活…Lock是对Synchronized的封装…

比如:

  1. 在控制锁的获取和释放,以及何时何地获取释放
  2. 使用Lock,可以很方便的实现锁的公平性ReentrantLock(boolean fair)
  3. 强大的API
    • 非阻塞获取锁:tryLock()
    • 可中断式获取线程acquireSharedInterruptibly(int arg)
    • 超时获取线程tryAcquireSharedNanos(int arg , long nanosTimeout)

一 . Conditon&Reentrantlock

1. Condition实现正确的通知/等待

  • 注意点,一定要在condition.wait()方法调用之前,使用lock.lock()方法获取对象同步锁,否则抛出异常
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
    public void waitMethod(){
        try{
            lock.lock();
            System.out.println(Thread.currentThread().getName()+"等待了..."+System.currentTimeMillis());
            try {
                condition.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }finally {
            lock.unlock();
        }

    }

    public void signalMethod(){
        try{
            lock.lock();
            System.out.println(Thread.currentThread().getName()+"被唤醒了..."+System.currentTimeMillis());
            condition.signal();
        }finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        demo1 demo1 = new demo1();

        new Thread(()->{
            demo1.waitMethod();
        }).start();

        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        demo1.signalMethod();
    }
}

运行结果:

1
2
Thread-0等待了...1549450097987
main被唤醒了...1549450099988

2 使用多个Condition实现,通知部分线程

接下来就是重头戏了Condition控制通知指定的线程醒来

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
public void waitMethod(){
    try{
        lock.lock();
        System.out.println(Thread.currentThread().getName()+"等待了..."+System.currentTimeMillis());
        try {
            condition.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }finally {
        lock.unlock();
    }

}

public void signalMethod(){
    try{
        lock.lock();
        System.out.println(Thread.currentThread().getName()+"被唤醒了..."+System.currentTimeMillis());
        condition.signal();
    }finally {
        lock.unlock();
    }
}

public static void main(String[] args) {
    demo1 demo1 = new demo1();

    new Thread(()->{
        demo1.waitMethod();
    }).start();

    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    demo1.signalMethod();
}
}

运行结果:

1
2
Thread-0等待了...1549450097987
main被唤醒了...1549450099988

2 使用多个Condition实现,通知部分线程

接下来就是重头戏了Condition控制通知指定的线程醒来

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

/*
* 使用多个Condition, 唤醒指定的线程
* */
public class demo2 {
private Lock lock = new ReentrantLock();
private Condition conditionA = lock.newCondition();
private Condition conditionB = lock.newCondition();

public void waitA(){
try{
 lock.lock();
 System.out.println(Thread.currentThread().getName()+"等待了..."+System.currentTimeMillis());
 try {
     conditionA.await();
     System.out.println(Thread.currentThread().getName()+"await之后的代码..."+System.currentTimeMillis());
 } catch (InterruptedException e) {
     e.printStackTrace();
 }
}finally {
 lock.unlock();
}
}
public void waitB(){
try{
    lock.lock();
    System.out.println(Thread.currentThread().getName()+"等待了..."+System.currentTimeMillis());
    try {
        conditionB.await();
        System.out.println(Thread.currentThread().getName()+"await之后的代码..."+System.currentTimeMillis());
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}finally {
    lock.unlock();
}
}

public void signalAALL(){
try{
 lock.lock();
 conditionA.signalAll();
 System.out.println(Thread.currentThread().getName()+" 执行唤醒操作. ."+System.currentTimeMillis());

}finally {
 lock.unlock();
}
}

public void signalBBLL(){
try{
    lock.lock();
    System.out.println(Thread.currentThread().getName()+"  被唤醒了.."+System.currentTimeMillis());
    conditionB.signalAll();
}finally {
    lock.unlock();
}
}

public static void main(String[] args) {
demo2 demo2 = new demo2();
ExecutorService executorService = Executors.newCachedThreadPool();
  executorService.execute(new Runnable() {
        @Override
        public void run() {
         demo2.waitA();
        // demo2.waitB();
        }
    });
try {
    Thread.sleep(2000);
} catch (InterruptedException e) {
    e.printStackTrace();
}
demo2.signalAALL();
System.out.println("mian 线程结束...");
}
}

结果:

1
2
3
4
pool-1-thread-1等待了...1549459953814
main 执行唤醒操作. .1549459955805
mian 线程结束...
pool-1-thread-1await之后的代码...1549459955805

  • 通过结果可以看到,ReentrantLock对象,可以唤醒指定种类的线程,使用个Condition让线程等待,就用哪个Condition去把它唤醒

公平锁与非公平锁

Lock锁分为公平锁和非公平锁,所谓公平锁,就是表示线程获取锁的顺序,是按照线程加锁的顺序来实现的,也就是FIFO的顺序先进先出的顺序,而非公平锁描述的则是一种锁的随机抢占机制,还可能会导致一些线程根本抢不着锁而被饿死,结果就是不公平了

  • ReentrantLock支持公平锁

ReentrantLock()
创建一个 ReentrantLock 的实例。

构造方法名 简介
ReentrantLock(boolean fair) 创建一个具有给定公平策略的 ReentrantLock。

ReentrantLock几组常用API

第一组:
方法名 | 作用
—|—
int getHoldCount() | 查询当前线程保存此锁的个数,(执行lock.lock()的次数),即重入的次数
int getQueueLength() | 返回正在等待获取此锁(调用该方法的锁)的线程的个数,比如有五条线程已经准备就绪了,其中一条线程首先执行lock.await()方法,紧接着调用该方法,获取到的返回值为4,说明有四条线程正在等待此lock
int getWaitQueueLength(Condition condition) | 返回正在等待和此锁相关的condition的线程数**

第二组:
方法名 | 作用
—|—
boolean hasQueuedThread(Thread thread) | 判断当前线程是否正在等到获取到此锁
boolean hasQueuedThreads() | 在所有的线程中查询,是否有线程正在等待此所的锁定
boolean hasWaiters(Condition condition) | 判断是否有线程正在等待和此condition相关的条件

第三组

方法名 作用
boolean isFair() 判断此锁是不是公平的
boolean isHeldByCurrentThread() 判断当前线程是否拿到了锁
boolean isLocked() 判断此锁是否由任意线程保持

第四组

lock()与lockInterruptibly()

假如出现下面几步,我们这两种加锁方法,会有什么反应?,第一: 线程A启动,使用lock()加锁,第二步: CPU的执行权被线程B抢到,且线程B使用interrupted()方法给线程A打上中断的标记..第三步: 线程A继续执行

  • 使用lock()加锁,假如我们没有使用isInterrupted()判断的话,代码会按照原来的顺序依次全部执行,没有异常,线程AB正常结束
  • 使用lockInterruptibly()加锁,线程A被中断后,会调用lockInterruptibly()报异常,进入catch代码块

tryLock()与Lock()

boolean tryLock()与void lock() 前者是有返回值的,两者都能去获取锁,而tryLock()直意尝试获取锁,有就拿,没有算了,它多了一步判断,它在调用时,会去判断此锁是否被其他线程锁定,如果没有,则获取,并返回ture

1
2
3
4
5
if(lock.tryLock()){
    //do something 1 ...
}else{
    //do something 2...
}

可以看到,使用tryLock(),不至于当前线程被阻塞住

方法名 作用
boolean tryLock(Long timeout,TimeUnit unit) 如果在给定的等待时间内,锁没有被其他线程拿到,并且当前线程也没有被中断,获取锁

await()和awaitUninterrupted()

当线程A await() 线程B给线程A打上中断的标记

  • 1.中断 2,抛出中断异常 3. 进入catch块

当线程A awaitUninterrupted() 线程B给线程A打上中断的标记

  • 1.中断

boolean awaitUtil(Date deadLine);

出现一下几种情况,线程被唤醒

  • 等待的时间结束
  • 在等待.被中断的过程被提前唤醒
  • 被中断

读写锁(排它锁,共享锁)

在一个不需要操作实例变量的方法中,完全可以使用读写锁来提升该方法的代码运行速度

  • 读操作相关的锁是共享锁,写操作相关的锁为排他锁

ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

  • 读读共享 lock.readLock().lock();
  • 写写互斥 lock.writeLock().lock();
  • 读写互斥

获取读写锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public void read() {
    lock.readLock().lock();
    System.out.println("获取到了读锁" + Thread.currentThread().getName() + "  " + System.currentTimeMillis());
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        lock.readLock().unlock();
    }
}

public void write(){
    try {
        lock.writeLock().lock();
        System.out.println(""+Thread.currentThread().getName()+"  " +System.currentTimeMillis());
        Thread.sleep(2000);
    }catch (Exception e){
        e.printStackTrace();
    }finally {
        lock.writeLock().unlock();
    }
}

创建四条线程测试

1
2
3
4
获取到了读锁Thread-0  1548851348805
获取到了读锁Thread-1  1548851348805
Thread-2  1548851350806
Thread-3  1548851352806

JDK8新增StampedLock对ReentrantReadWriteLock进行增强

stamped:贴上邮票的; 盖上邮戳的,拿到锁之后会返回给我们一个票据,根据这个Stamp的值判断是在读的时候发生了写,还是在写的时候发生了读操作

解决的问题:
在高并发的情况下,读操作的次数远远大于写操作,,因为读写互斥,写操作可能就会出现饥饿的情况,一直抢占不到cpu的资源

解决方法:

  1. 当然可以使用公平的ReadWriteLock,但是依然有性能问题

  2. StampedLock的乐观锁实现了读写共享提升了!

StampedLock里面有两种锁
乐观锁:

读锁并不会阻塞写锁 

1
public long tryOptimisticRead() {...}

悲观锁:

读写互斥,和ReentrantReadWriteLock实现相同的功能

API:

独占的获取写锁,若锁被其他线程获取到,则阻塞,注意它是相对于ReentrantReadWriteLock来讲,它是有返回值的,返回值的作用:

  • 释放锁(unlock的需要参数)
  • 进行锁的转换
1
2
3
4
5
6
7
8
9
10
11
12
/**
  * Exclusively acquires the lock, blocking if necessary
  * until available.
  *
  * @return a stamp that can be used to unlock or convert mode
  */
 public long writeLock() {
     long s, next;  // bypass acquireWrite in fully unlocked case only
     return ((((s = state) & ABITS) == 0L &&
              U.compareAndSwapLong(this, STATE, s, next = s + WBIT)) ?
             next : acquireWrite(false, 0L));
 }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

//一旦写锁可用立即获取,返回值可以用作释放锁或被锁的转换使用, 0表示没有获取到锁
 /**
 * Exclusively acquires the lock if it is immediately available.
 *
 * @return a stamp that can be used to unlock or convert mode,
 * or zero if the lock is not available
 */
public long tryWriteLock() {
    long s, next;
    return ((((s = state) & ABITS) == 0L &&
             U.compareAndSwapLong(this, STATE, s, next = s + WBIT)) ?
            next : 0L);
}


//延迟获取锁
     /**
 * Exclusively acquires the lock if it is available within the
 * given time and the current thread has not been interrupted.
 * Behavior under timeout and interruption matches that specified
 * for method {@link Lock#tryLock(long,TimeUnit)}.
 *
 * @param time the maximum time to wait for the lock
 * @param unit the time unit of the {@code time} argument
 * @return a stamp that can be used to unlock or convert mode,
 * or zero if the lock is not available
 * @throws InterruptedException if the current thread is interrupted
 * before acquiring the lock
 */
public long tryWriteLock(long time, TimeUnit unit)


...


// 非独占的获取读锁
  /**
 * Non-exclusively acquires the lock, blocking if necessary
 * until available.
 *
 * @return a stamp that can be used to unlock or convert mode
 */
public long readLock() {
    long s = state, next;  // bypass acquireRead on common uncontended case
    return ((whead == wtail && (s & ABITS) < RFULL &&
             U.compareAndSwapLong(this, STATE, s, next = s + RUNIT)) ?
            next : acquireRead(false, 0L));
}


// 一旦锁可用,立即非独占的获取读锁
/**
 * Non-exclusively acquires the lock if it is immediately available.
 *
 * @return a stamp that can be used to unlock or convert mode,
 * or zero if the lock is not available
 */
public long tryReadLock() {
    for (;;) {
        long s, m, next;
        if ((m = (s = state) & ABITS) == WBIT)
            return 0L;
        else if (m < RFULL) {
            if (U.compareAndSwapLong(this, STATE, s, next = s + RUNIT))
                return next;
        }
        else if ((next = tryIncReaderOverflow(s)) != 0L)
            return next;
    }
}

乐观锁!!!它获取到的锁,读写锁非互斥
返回一个标记,这个标记过一会用去 校验, 如果锁是排它锁,返回零

1
2
3
4
5
6
7
8
9
10
11

/**
 * Returns a stamp that can later be validated, or zero
 * if exclusively locked.
 *
 * @return a stamp, or zero if exclusively locked
 */
public long tryOptimisticRead() {
    long s;
    return (((s = state) & WBIT) == 0L) ? (s & SBITS) : 0L;
}

校验,如果锁还没被任何线程获取,获取被持有当前stamp的线程获取返回true , 如果 stamp为0,返回false

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
 /**
 * Returns true if the lock has not been exclusively acquired
 * since issuance of the given stamp. Always returns false if the
 * stamp is zero. Always returns true if the stamp represents a
 * currently held lock. Invoking this method with a value not
 * obtained from {@link #tryOptimisticRead} or a locking method
 * for this lock has no defined effect or result.
 *
 * @param stamp a stamp
 * @return {@code true} if the lock has not been exclusively acquired
 * since issuance of the given stamp; else false
 */
public boolean validate(long stamp) {
    U.loadFence();
    return (stamp & SBITS) == (state & SBITS);
}

锁的释放

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
  
  /**
   * If the lock state matches the given stamp, releases the
   * exclusive lock.
   *
   * @param stamp a stamp returned by a write-lock operation
   * @throws IllegalMonitorStateException if the stamp does
   * not match the current state of this lock
   */
  public void unlockWrite(long stamp) {
      WNode h;
      if (state != stamp || (stamp & WBIT) == 0L)
          throw new IllegalMonitorStateException();
      state = (stamp += WBIT) == 0L ? ORIGIN : stamp;
      if ((h = whead) != null && h.status != 0)
          release(h);
  }
  
//释放任意匹配成功的锁
  /**
   * If the lock state matches the given stamp, releases the
   * corresponding mode of the lock.
   *
   * @param stamp a stamp returned by a lock operation
   * @throws IllegalMonitorStateException if the stamp does
   * not match the current state of this lock
   */
  public void unlock(long stamp) {

读,写 锁之间的转换

1
2
3
4
5
6
7
8
9
10
 public long tryConvertToWriteLock(long stamp) {
 ..}
 
  public long tryConvertToWriteLock(long stamp) {
..}
//释放读锁
public void unlockRead(long stamp) {..}

//释放写锁
public void unlockWrite(long stamp) {..}

简单使用后

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
/*
* StampedLock的简单使用
* */
public class StampedLock01 {

private  int balance;
StampedLock stamptedLock = new StampedLock();

//悲观读
public void read(){
    long s = stamptedLock.readLock();
try{
    try {
        System.out.println("拿到读锁"+Thread.currentThread().getName()+System.currentTimeMillis());
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}finally {
    stamptedLock.unlockRead(s);
}
}

// 乐观锁
public void OptimismRead(){
    //获取乐观锁,拿到 标记
    long stm = stamptedLock.tryOptimisticRead();
    try {
        System.out.println("一开始读取到的balance=="+balance);

        //方便测试     睡一会
        Thread.sleep(200);
        // 在读的时候,可能会出现现写操作-- 判断
        if(!stamptedLock.validate(stm)){  //1 锁空闲可用  2 拥有当前stm的线程获取到锁,返回true  其他返回false
            //重新读取
            long l = stamptedLock.readLock();
            System.out.println("乐观锁中发现了在读时,发现写操作,重新读结果为:  "+balance);
            // 更新标记, 用于锁的释放
            stm=l;
        }
    }catch (Exception e){
        e.printStackTrace();
    }finally {
        //释放锁
        stamptedLock.unlockRead(stm);

    }
}

// 带条件的读写锁
public void MyConditionReadWriteLock(int v){
    //判断balance是否符合更新的条件
    long stm = stamptedLock.readLock();
    try{

        // 为什么不会if  而用while
        while(stm>0){
            //转换成写锁;
            long s1 = stamptedLock.tryConvertToWriteLock(stm);
            if (s1!=0){// 成功转换
                balance+=v;
                System.out.println("进行写操作:  balance=="+balance+99);
                stm=s1;
                break;
            }else{  //没有转换成功
                //释放读锁
                stamptedLock.unlockRead(stm);
                // 获取写
                long s2 = stamptedLock.writeLock();
                stm=s2;
                System.out.println("手动获取写锁...");
            }
        }
    }finally {
        stamptedLock.unlock(stm);
    }


}



//独占的写
public void write(int v){
        long stamp = stamptedLock.writeLock();
        try{
            balance+=v;
            System.out.println("进行写的操作...结果:  "+balance+"  "+Thread.currentThread().getName()+"  "+"write中的标记为:"+stamp);
            Thread.sleep(2000);
        }catch (Exception e){
            e.printStackTrace();
        }
        finally {
            stamptedLock.unlockWrite(stamp);
        }
}


public static void main(String[] args) {

    StampedLock01 stampedLock01 = new StampedLock01();

    //测试普通的read  // 测试成功 异步执行

    //测试独占的写  成功
/*       new Thread(()->{
        // 乐观读
        stampedLock01.OptimismRead();
    }).start();

    new Thread(()->{
        stampedLock01.write(3);
    }).start();
*/
    new Thread(()->{
        stampedLock01.MyConditionReadWriteLock(1);
    }).start();
}

}

最后提一下 > 如何选择大名鼎鼎的AQS最有名的实现类ReentrantLock和synchronized这个JVM提供的锁,一开始synchronized是一个笨重的重量级锁,但是jdk1.5之后,进行了偏向锁,轻量级锁的优化,使它的性能和ReentrantLock擦不多了,于是没有特殊的要求,官方推荐使用synchronized
什么情况下使用ReentrantLock呢?

  • 使用它特有的公平锁
  • 使用它的Condition类,分组唤醒指定的线程
  • 提供了能够中断正在等待锁的线程的机制,lock.lockInterrupted()